package com.swak.frame.hystrix.stream;

import java.io.IOException;
import java.io.PrintWriter;
import java.util.ResourceBundle;
import java.util.concurrent.atomic.AtomicBoolean;

import javax.servlet.ServletException;
import javax.servlet.http.HttpServletRequest;
import javax.servlet.http.HttpServletResponse;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import rx.Observable;
import rx.Subscriber;
import rx.Subscription;
import rx.schedulers.Schedulers;

public abstract class HystrixSampleSseStream {
	protected final Observable<String> sampleStream;

	private static final Logger logger = LoggerFactory.getLogger(HystrixSampleSseStream.class);

	// wake up occasionally and check that poller is still alive. this value
	// controls how often
	protected static final int DEFAULT_PAUSE_POLLER_THREAD_DELAY_IN_MS = 500;

	private static ResourceBundle lStrings = ResourceBundle.getBundle("javax.servlet.http.LocalStrings");

	private final int pausePollerThreadDelayInMs;

	/*
	 * Set to true upon shutdown, so it's OK to be shared among all
	 * SampleSseServlets
	 */
	private static volatile boolean isDestroyed = false;

	protected HystrixSampleSseStream(Observable<String> sampleStream) {
		this.sampleStream = sampleStream;
		this.pausePollerThreadDelayInMs = DEFAULT_PAUSE_POLLER_THREAD_DELAY_IN_MS;
	}

	protected HystrixSampleSseStream(Observable<String> sampleStream, int pausePollerThreadDelayInMs) {
		this.sampleStream = sampleStream;
		this.pausePollerThreadDelayInMs = pausePollerThreadDelayInMs;
	}

	protected abstract int getMaxNumberConcurrentConnectionsAllowed();

	protected abstract int getNumberCurrentConnections();

	protected abstract int incrementAndGetCurrentConcurrentConnections();

	protected abstract void decrementCurrentConcurrentConnections();

	/**
	 * Handle incoming GETs
	 */

	public void doGet(HttpServletRequest request, HttpServletResponse response) throws ServletException, IOException {
		if (isDestroyed) {
			response.sendError(503, "Service has been shut down.");
		} else {
			handleRequest(request, response);
		}
	}

	public void doPost(HttpServletRequest req, HttpServletResponse resp) throws ServletException, IOException {
		String protocol = req.getProtocol();
		String msg = lStrings.getString("http.method_post_not_supported");
		if (protocol.endsWith("1.1")) {
			resp.sendError(405, msg);
		} else {
			resp.sendError(400, msg);
		}
	}

	/**
	 * WebSphere won't shutdown a servlet until after a 60 second timeout if there
	 * is an instance of the servlet executing a request. Add this method to enable
	 * a hook to notify Hystrix to shutdown. You must invoke this method at
	 * shutdown, perhaps from some other servlet's destroy() method.
	 */
	public static void shutdown() {
		isDestroyed = true;
	}

	public void init() throws ServletException {
		isDestroyed = false;
	}

	/**
	 * Handle servlet being undeployed by gracefully releasing connections so poller
	 * threads stop.
	 */

	public void destroy() {
		/* set marker so the loops can break out */
		isDestroyed = true;
	}

	/**
	 * - maintain an open connection with the client - on initial connection send
	 * latest data of each requested event type - subsequently send all changes for
	 * each requested event type
	 *
	 * @param request
	 *            incoming HTTP Request
	 * @param response
	 *            outgoing HTTP Response (as a streaming response)
	 * @throws javax.servlet.ServletException
	 * @throws java.io.IOException
	 */
	private void handleRequest(HttpServletRequest request, final HttpServletResponse response)
			throws ServletException, IOException {
		final AtomicBoolean moreDataWillBeSent = new AtomicBoolean(true);
		Subscription sampleSubscription = null;

		/* ensure we aren't allowing more connections than we want */
		int numberConnections = incrementAndGetCurrentConcurrentConnections();
		try {
			int maxNumberConnectionsAllowed = getMaxNumberConcurrentConnectionsAllowed(); // may change at runtime, so
																							// look this up for each
																							// request
			if (numberConnections > maxNumberConnectionsAllowed) {
				response.sendError(503, "MaxConcurrentConnections reached: " + maxNumberConnectionsAllowed);
			} else {
				/* initialize response */
				response.setHeader("Content-Type", "text/event-stream;charset=UTF-8");
				response.setHeader("Cache-Control", "no-cache, no-store, max-age=0, must-revalidate");
				response.setHeader("Pragma", "no-cache");

				final PrintWriter writer = response.getWriter();

				// since the sample stream is based on Observable.interval, events will get
				// published on an RxComputation thread
				// since writing to the servlet response is blocking, use the Rx IO thread for
				// the write that occurs in the onNext
				sampleSubscription = sampleStream.observeOn(Schedulers.io()).subscribe(new Subscriber<String>() {
					@Override
					public void onCompleted() {
						logger.error("HystrixSampleSseServlet: ({}) received unexpected OnCompleted from sample stream",
								getClass().getSimpleName());
						moreDataWillBeSent.set(false);
					}

					@Override
					public void onError(Throwable e) {
						moreDataWillBeSent.set(false);
					}

					@Override
					public void onNext(String sampleDataAsString) {
						if (sampleDataAsString != null) {
							writer.print("data: " + sampleDataAsString + "\n\n");
							// explicitly check for client disconnect - PrintWriter does not throw
							// exceptions
							if (writer.checkError()) {
								moreDataWillBeSent.set(false);
							}
							writer.flush();
						}
					}
				});

				while (moreDataWillBeSent.get() && !isDestroyed) {
					try {
						Thread.sleep(pausePollerThreadDelayInMs);
						// in case stream has not started emitting yet, catch any clients which
						// connect/disconnect before emits start
						writer.print("ping: \n\n");
						// explicitly check for client disconnect - PrintWriter does not throw
						// exceptions
						if (writer.checkError()) {
							moreDataWillBeSent.set(false);
						}
						writer.flush();
					} catch (InterruptedException e) {
						moreDataWillBeSent.set(false);
					}
				}
			}
		} finally {
			decrementCurrentConcurrentConnections();
			if (sampleSubscription != null && !sampleSubscription.isUnsubscribed()) {
				sampleSubscription.unsubscribe();
			}
		}
	}
}
